package com.fudan.run.action;

import java.util.Map;

import org.apache.spark.api.java.JavaRDD;
import org.elasticsearch.spark.rdd.api.java.JavaEsSpark;

import com.fudan.cfg.RunConfiguration;
import com.fudan.cfg.action.SaveToEs;
import com.fudan.cfg.base.ActionDef;
import com.fudan.run.VariableRddMap;
import com.fudan.run.ctx.annotation.ActionAdapter;

@ActionAdapter("saveToEs")
public class SaveToEsAction extends ActionHandler<SaveToEs>{

	@Override
	public void action(SaveToEs actionNode, VariableRddMap variableRddMap,RunConfiguration runConfiguration) {

		JavaRDD<Map<String, Object>> targetRdd = variableRddMap.getRdd(actionNode.getTarget()).repartition(actionNode.getPartition());

		JavaEsSpark.saveToEs(targetRdd, actionNode.getIndex());

	}

}
